package sample.cluster.trans2;

import akka.actor.ActorRef;
import akka.actor.Terminated;
import akka.actor.UntypedActor;
import sample.cluster.transformation.TransformationMessages.JobFailed;
import sample.cluster.transformation.TransformationMessages.TransformationJob;
import sample.cluster.transformation.TransformationMessages.TransformationResult;
import scala.concurrent.ExecutionContext;
import scala.concurrent.duration.Duration;
import scala.concurrent.duration.FiniteDuration;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import static sample.cluster.transformation.TransformationMessages.BACKEND_REGISTRATION;

//#frontend
public class TransformationFrontend extends UntypedActor {

    List<ActorRef> backends = new ArrayList<ActorRef>();
    int jobCounter = 0;
    private ActorRef otherActorRef;

    @Override
    public void onReceive(Object message) {
        if(message.equals("start")){
            otherActorRef = getSender();
//            getContext().watch(getSender());
            final FiniteDuration interval = Duration.create(2, TimeUnit.SECONDS);
            final ExecutionContext ec = getContext().system().dispatcher();
            getContext().system().scheduler().schedule(interval, interval, new Runnable() {
                public void run() {
//                    if(otherActorRef!=null) {
                        System.out.println("=======================>"+otherActorRef.isTerminated());
                    if(!otherActorRef.isTerminated()) {
                        otherActorRef.tell("from fronted", ActorRef.noSender());

                    }
                }
            }, ec);
        }

        if (message instanceof TransformationResult) {
            System.out.println("I am here " + message);
        }

        if ((message instanceof TransformationJob) && backends.isEmpty()) {
            TransformationJob job = (TransformationJob) message;
            getSender().tell(
                    new JobFailed("Service unavailable, try again later", job),
                    getSender());

        } else if (message instanceof TransformationJob) {
            TransformationJob job = (TransformationJob) message;
            jobCounter++;
            backends.get(jobCounter % backends.size())
                    .forward(job, getContext());

        } else if (message.equals(BACKEND_REGISTRATION)) {
            getContext().watch(getSender());
            backends.add(getSender());

        } else if (message instanceof Terminated) {
            Terminated terminated = (Terminated) message;
            backends.remove(terminated.getActor());

        } else {
            unhandled(message);
        }
    }

}
//#frontend
